package main

import (
	"encoding/json"
	"fmt"
	"io"
	"os"
	"sync"
	"sync/atomic"
	"time"

	"go.nanomsg.org/mangos/v3"
	"go.nanomsg.org/mangos/v3/protocol/pub"
	"go.nanomsg.org/mangos/v3/protocol/sub"
	"go.nanomsg.org/mangos/v3/protocol/rep"
	"go.nanomsg.org/mangos/v3/protocol/req"

	_ "go.nanomsg.org/mangos/v3/transport/tcp"
)

const (
	heartbeatInterval  = 2 * time.Second // 心跳间隔
	detectionWindow    = 3               // 3个心跳周期未收到心跳则切换 Master
	modeSwitchTimeout  = 5 * time.Second // 模式切换超时时间
	ackWaitTimeout     = 1 * time.Second // ACK 等待超时
)

type Mode int

const (
	ModeListening Mode = iota
	ModeMaster
	ModeSlave
)

type keepAliveConfig struct {
	IPAddress string `json:"ip_address"`
	Port      int    `json:"port"`
	PeerIP    string `json:"peer_ip"`
	SyncPort  int    `json:"sync_port"`
	SecretKey string `json:"secret_key"`
	Enable    bool   `json:"enable"`
}

type Message struct {
	Type    string      `json:"type"`
	Content interface{} `json:"content"`
}

var (
	keepAlive  keepAliveConfig
	mode       Mode = ModeListening
	mu         sync.RWMutex
	metrics    struct {
		heartbeatsSent     int64
		heartbeatsReceived int64
	}
	sockPub, sockSub, sockAck mangos.Socket
)

// 加载配置文件
func loadConfig() error {
	file, err := os.Open("config.json")
	if err != nil {
		return err
	}
	defer file.Close()

	bytes, err := io.ReadAll(file)
	if err != nil {
		return err
	}

	err = json.Unmarshal(bytes, &keepAlive)
	return err
}

// 关闭所有 socket，避免端口占用
func closeSockets() {
	if sockPub != nil {
		sockPub.Close()
	}
	if sockSub != nil {
		sockSub.Close()
	}
	if sockAck != nil {
		sockAck.Close()
	}
}

// 模式切换逻辑
func startModeSwitch() {
	fmt.Println("启动模式切换逻辑, 等待5秒...")
	timer := time.NewTimer(modeSwitchTimeout)
	defer timer.Stop()

	for {
		select {
		case <-timer.C:
			mu.Lock()
			if atomic.LoadInt64(&metrics.heartbeatsReceived) == 0 {
				mode = ModeMaster
				go startMasterMode()
			} else {
				mode = ModeSlave
				go startSlaveMode()
			}
			mu.Unlock()
			return
		}
	}
}

// Master 模式
func startMasterMode() {
	fmt.Println("切换到 Master 模式...")
	closeSockets() // 避免端口重复监听
	go startHeartbeat()
}

// Slave 模式
func startSlaveMode() {
	fmt.Println("切换到 Slave 模式...")
	closeSockets() // 避免端口重复监听
	go startListener()
	go startAckServer()
}

// Master 发送心跳
func startHeartbeat() {
	var err error
	sockPub, err = pub.NewSocket()
	if err != nil {
		fmt.Println("创建 PUB Socket 失败:", err)
		return
	}
	defer sockPub.Close()

	addr := fmt.Sprintf("tcp://%s:%d", keepAlive.IPAddress, keepAlive.Port)
	if err := sockPub.Listen(addr); err != nil {
		fmt.Println("PUB 监听失败:", err)
		return
	}

	fmt.Println("Master 启动，发布心跳数据...")
	for {
		message := Message{Type: "heartbeat", Content: ""}
		data, _ := json.Marshal(message)

		err := sockPub.Send(data)
		if err != nil {
			fmt.Println("心跳发送失败:", err)
		} else {
			atomic.AddInt64(&metrics.heartbeatsSent, 1)
			fmt.Println("发送心跳数据")
		}

		time.Sleep(heartbeatInterval)
	}
}

// Slave 监听心跳
func startListener() {
	var err error
	sockSub, err = sub.NewSocket()
	if err != nil {
		fmt.Println("创建 SUB Socket 失败:", err)
		return
	}
	defer sockSub.Close()

	addr := fmt.Sprintf("tcp://%s:%d", keepAlive.PeerIP, keepAlive.Port)
	if err := sockSub.Dial(addr); err != nil {
		fmt.Println("SUB 连接失败:", err)
		return
	}

	_ = sockSub.SetOption(mangos.OptionSubscribe, "")

	fmt.Println("Slave 启动，监听心跳数据...")
	missedHeartbeats := 0

	for {
		data, err := sockSub.Recv()
		if err != nil {
			fmt.Println("接收心跳失败:", err)
			missedHeartbeats++
		} else {
			var message Message
			if err := json.Unmarshal(data, &message); err != nil {
				fmt.Println("解析心跳消息失败:", err)
				continue
			}

			if message.Type == "heartbeat" {
				atomic.AddInt64(&metrics.heartbeatsReceived, 1)
				fmt.Println("收到心跳信号，发送 ACK")
				go sendAck()
				missedHeartbeats = 0
			}
		}

		if missedHeartbeats >= detectionWindow {
			fmt.Println("超过检测窗口未收到心跳，切换为 Master")
			mu.Lock()
			mode = ModeMaster
			mu.Unlock()
			go startMasterMode()
			return
		}
	}
}

// 发送 ACK 到 Master
func sendAck() {
	sockReq, err := req.NewSocket()
	if err != nil {
		fmt.Println("创建 ACK 请求 Socket 失败:", err)
		return
	}
	defer sockReq.Close()

	addr := fmt.Sprintf("tcp://%s:%d", keepAlive.PeerIP, keepAlive.SyncPort)
	if err := sockReq.Dial(addr); err != nil {
		fmt.Println("ACK 连接失败:", err)
		return
	}

	message := Message{Type: "ack", Content: ""}
	data, _ := json.Marshal(message)
	err = sockReq.Send(data)
	if err != nil {
		fmt.Println("ACK 发送失败:", err)
	}
}

// Master 监听 ACK
func startAckServer() {
	var err error
	sockAck, err = rep.NewSocket()
	if err != nil {
		fmt.Println("创建 ACK 监听 Socket 失败:", err)
		return
	}
	defer sockAck.Close()

	addr := fmt.Sprintf("tcp://%s:%d", keepAlive.IPAddress, keepAlive.SyncPort)
	if err := sockAck.Listen(addr); err != nil {
		fmt.Println("ACK 监听失败:", err)
		return
	}

	fmt.Println("Master 监听 ACK...")
	for {
		data, err := sockAck.Recv()
		if err != nil {
			fmt.Println("ACK 接收失败:", err)
			continue
		}

		var message Message
		if err := json.Unmarshal(data, &message); err != nil {
			fmt.Println("解析 ACK 失败:", err)
			continue
		}

		if message.Type == "ack" {
			fmt.Println("收到 ACK")
		}
	}
}

func main() {
	err := loadConfig()
	if err != nil {
		fmt.Println("加载配置文件失败:", err)
		return
	}

	go startListener()
	go startModeSwitch()

	select {}
}
